package cn.qdhlm.frontendprocessor.common.config;

import cn.qdhlm.frontendprocessor.core.callback.RocketMQMessageCallback;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Name:RocketMqConfig
 * @description:
 * @author: xinY
 **/
@Configuration
public class RocketMqConfig {
    @Value("${rocketmq.producer.group}")
    private String producerGroup;
    @Value("${rocketmq.name-server}")
    private String rmqNameSrvAddr;
    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;
    private static final String TOPICS = "Command";
    @Autowired
    private RocketMQMessageCallback messageCallback;

    @Bean
    public DefaultMQProducer getProducer(){
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setVipChannelEnabled(false);
        producer.setNamesrvAddr(rmqNameSrvAddr);

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return producer;
    }

    @Bean
    public DefaultMQPushConsumer getConsumer(){
        // 获取consumer实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

        // 连接namesrv
        consumer.setNamesrvAddr(rmqNameSrvAddr);

        // 关闭vip通道 端口是listenport -2
        consumer.setVipChannelEnabled(false);

        // CONSUME_FROM_LAST_OFFSET不要历史（CONSUME_FROM_FIRST_OFFSET）包括历史
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        // 集群消费（广播消费）
        consumer.setMessageModel(MessageModel.CLUSTERING);

        // 注册消息监听器
        consumer.registerMessageListener(messageCallback);

        try {
            // 订阅

            consumer.subscribe(TOPICS,"*");
            // 启动
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return consumer;

    }
}
